/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.urick.notify.nacos.manager;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.urick.notify.nacos.event.Event;
import com.urick.notify.nacos.pub.EventPublisherFactory;
import com.urick.notify.nacos.pub.StandardPublisher;
import com.urick.notify.nacos.pub.IPublisher;
import com.urick.notify.nacos.sub.Subscriber;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;

/**
 * Unified Event Notify Center.
 * 消息通知中心，负责处理消息调度
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
public class NotifyManager {

    private static final Log log = LogFactory.get();

    public static int ringBufferSize = 16384;

    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    private static final EventPublisherFactory DEFAULT_PUBLISHER_FACTORY;

    private static final NotifyManager INSTANCE = new NotifyManager();

    /**
     * Publisher management container.
     */
    private final Map<String, IPublisher> publisherMap = new ConcurrentHashMap<>(16);

    static {
        DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
            try {
                IPublisher publisher = new StandardPublisher();
                publisher.init(cls, buffer);
                return publisher;
            } catch (Throwable ex) {
                log.error("Service class newInstance has error : ", ex);
                throw new RuntimeException(ex);
            }
        };
        Runtime.getRuntime().addShutdownHook(new Thread(NotifyManager::shutdown));
    }


    public static Map<String, IPublisher> getPublisherMap() {
        return INSTANCE.publisherMap;
    }

    public static IPublisher getPublisher(Class<? extends Event> topic) {
        return INSTANCE.publisherMap.get(topic.getCanonicalName());
    }

    /**
     * Shutdown the several publisher instance which notify center has.
     */
    public static void shutdown() {
        if (!CLOSED.compareAndSet(false, true)) {
            return;
        }
        log.warn("[NotifyCenter] Start destroying Publisher");

        for (Map.Entry<String, IPublisher> entry : getPublisherMap().entrySet()) {
            try {
                IPublisher eventPublisher = entry.getValue();
                eventPublisher.shutdown();
            } catch (Throwable e) {
                log.error("[EventPublisher] shutdown has error : ", e);
            }
        }
        log.warn("[NotifyCenter] Destruction of the end");
    }

    /**
     * Register a Subscriber. If the Publisher concerned by the Subscriber does not exist, then PublihserMap will
     * preempt a placeholder Publisher with default EventPublisherFactory first.
     *
     * @param consumer subscriber
     */
    public static void registerSubscriber(final Subscriber consumer) {
        registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
    }

    /**
     * Register a Subscriber. If the Publisher concerned by the Subscriber does not exist, then PublihserMap will
     * preempt a placeholder Publisher with specified EventPublisherFactory first.
     *
     * @param consumer subscriber
     * @param factory  publisher factory.
     */
    public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
        addSubscriber(consumer, factory);
    }

    /**
     * Add a subscriber to publisher.
     *
     * @param consumer subscriber instance.
     * @param factory  publisher factory.
     */
    private static void addSubscriber(final Subscriber consumer, EventPublisherFactory factory) {
        Class<? extends Event> subscribeType = consumer.subscribeType();
        final String topic = subscribeType.getCanonicalName();
        synchronized (NotifyManager.class) {
            //   computeIfAbsent is a unsafe method.
            computeIfAbsent(getPublisherMap(), topic, factory, subscribeType, ringBufferSize);
        }
        IPublisher publisher = getPublisherMap().get(topic);
        publisher.addSubscriber(consumer);
    }


    /**
     * Request publisher publish event Publishers load lazily, calling publisher. Start () only when the event is
     * actually published.
     *
     * @param event class Instances of the event.
     */
    public static boolean publishEvent(final Event event) {
        try {
            return publishEvent(event.getClass(), event);
        } catch (Throwable ex) {
            log.error("There was an exception to the message publishing : ", ex);
            return false;
        }
    }

    /**
     * Request publisher publish event Publishers load lazily, calling publisher.
     *
     * @param eventType class Instances type of the event type.
     * @param event     event instance.
     */
    private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
        final String topic = eventType.getCanonicalName();
        IPublisher publisher = getPublisher(eventType);
        if (publisher != null) {
            return publisher.publish(event);
        }
        log.warn("There are no [{}] publishers for this event, please register", topic);
        return false;
    }

    /**
     * Register publisher with default factory.
     *
     * @param eventType    class Instances type of the event type.
     * @param queueMaxSize the publisher's queue max size.
     */
    public static IPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {
        return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);
    }

    /**
     * Register publisher with specified factory.
     *
     * @param eventType    class Instances type of the event type.
     * @param factory      publisher factory.
     * @param queueMaxSize the publisher's queue max size.
     */
    public static IPublisher registerToPublisher(final Class<? extends Event> eventType, final EventPublisherFactory factory, final int queueMaxSize) {

        final String topic = eventType.getCanonicalName();
        synchronized (NotifyManager.class) {
            //   computeIfAbsent is a unsafe method.
            computeIfAbsent(getPublisherMap(), topic, factory, eventType, queueMaxSize);
        }
        return getPublisherMap().get(topic);
    }

    /**
     * Register publisher.
     *
     * @param eventType class Instances type of the event type.
     * @param publisher the specified event publisher
     */
    public static void registerToPublisher(final Class<? extends Event> eventType, final IPublisher publisher) {
        if (null == publisher) {
            return;
        }
        final String topic = eventType.getCanonicalName();
        synchronized (NotifyManager.class) {
            getPublisherMap().putIfAbsent(topic, publisher);
        }
    }

    /**
     * Deregister publisher.
     *
     * @param eventType class Instances type of the event type.
     */
    public static void deregisterPublisher(final Class<? extends Event> eventType) {
        final String topic = eventType.getCanonicalName();
        IPublisher publisher = getPublisherMap().remove(topic);
        try {
            publisher.shutdown();
        } catch (Throwable ex) {
            log.error("There was an exception when publisher shutdown : ", ex);
        }
    }

    public static <K, C, V, T> void computeIfAbsent(Map<K, V> target, K key, BiFunction<C, T, V> mappingFunction, C param1, T param2) {
        Objects.requireNonNull(target, "target");
        Objects.requireNonNull(key, "key");
        Objects.requireNonNull(mappingFunction, "mappingFunction");
        Objects.requireNonNull(param1, "param1");
        Objects.requireNonNull(param2, "param2");
        V val = target.get(key);
        if (val == null) {
            V ret = mappingFunction.apply(param1, param2);
            target.put(key, ret);
        }
    }
}
